package com.zengqingfa.rocketmq.consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.UUID;

/**
 *
 * @fileName: MqDemo01Consumer
 * @author: zengqf3
 * @date: 2021-4-2 22:36
 * @description:
 */
@Configuration
@Slf4j
public class MqDemo01Consumer {
    @Value("${rocketmq.namesrvAddr}")
    private String namesrvAddr;

    @Value("${rocketmq.topic.demo01}")
    private String topic;

    @Value("${rocketmq.tag.demo01}")
    private String tag;

    @Autowired
    private MqDemo01Listener mqDemo01Listener;

    @Bean
    @ConditionalOnMissingBean
    public DefaultMQPushConsumer defaultMQPushConsumer() throws RuntimeException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(UUID.randomUUID().toString());
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.setConsumeThreadMin(10);
        consumer.setConsumeThreadMax(20);

        // 设置 consumer 第一次启动是从队列头部开始消费还是队列尾部开始消费
        // 如果非第一次启动，那么按照上次消费的位置继续消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        // 设置消费模型，集群还是广播，默认为集群
        consumer.setMessageModel(MessageModel.CLUSTERING);
        // 设置一次消费消息的条数，默认为 1 条
        consumer.setConsumeMessageBatchMaxSize(1);
        consumer.registerMessageListener(mqDemo01Listener);
        try {
            // 设置该消费者订阅的主题和tag，如果是订阅该主题下的所有tag，使用*；
            consumer.subscribe(topic, tag);
            // 启动消费
            consumer.start();
            log.info("consumer is started. groupName:{}, topic:{}, tag:{},namesrvAddr:{}", consumer.getConsumerGroup(),
                    topic, tag, namesrvAddr);
        } catch (Exception e) {
            log.error("failed to start consumer . groupName:{}, topic:{},tag:{}, namesrvAddr:{}",
                    consumer.getConsumerGroup(), topic, tag, namesrvAddr, e);
            throw new RuntimeException(e);
        }
        return consumer;
    }
}
